Defining and Visualizing fmdtools Model Structures

To ensure that a simulation meets the intent of a modeller, it is important to carefully define the structure of the model and run order. This notebook will demonstrate fmdtools’ interfaces both for setting up model structures and for fault visualization.

NOTE: For some of these visualizations to display properly without re-running code use, File -> Trust Notebook.

from fmdtools.define.architecture.function import FunctionArchitecture
from fmdtools.define.block.function import Function
from fmdtools.define.block.component import Component
import fmdtools.analyze as an
import fmdtools.sim.propagate as prop

Basics

An fmdtools model is made up of functions–model structures with behavioral methods and internal states–and flows–data relationships between functions. These functions and flows are defined in python classes and thus may be instantiated multiple times in a model to create multi-component models with complex interactions. The structure of these model classes is shown below:

Creating a model thus involves: - defining the function and flow classes defining the behavior of the model in the model module - defining the specific structure for the model: function and flow objects (instantiations of functions) and their relationships.

Model structure visualization is performed in the an.graph.FunctionArchitectureGraph class and its sub-classes.

help(an.graph.FunctionArchitectureGraph)

Help on class FunctionArchitectureGraph in module fmdtools.analyze.graph:

class FunctionArchitectureGraph(Graph)
 |  FunctionArchitectureGraph(mdl, get_states=True, time=0.0, **kwargs)
 |
 |  Creates a Graph of Model functions and flow for display, where both functions
 |  and flows are nodes.
 |
 |  If get_states option is used on instantiation, a `states` dict is associated
 |  with the edges/nodes which can then be used to visualize function/flow attributes.
 |
 |  Method resolution order:
 |      FunctionArchitectureGraph
 |      Graph
 |      builtins.object
 |
 |  Methods defined here:
 |
 |  __init__(self, mdl, get_states=True, time=0.0, **kwargs)
 |      Generate the FunctionArchitectureGraph corresponding to a given Model.
 |
 |      Parameters
 |      ----------
 |      mdl : define.Model
 |          fmdtools model to represent graphically
 |      get_states : bool, optional
 |          Whether to copy states to the node/edge 'states' property.
 |          The default is True.
 |      time: float
 |          Time model is run at (to execute indicators at). Default is 0.0
 |      **kwargs : kwargs
 |          (placeholder for kwargs)
 |
 |  draw_graphviz(self, layout='twopi', overlap='voronoi', **kwargs)
 |      Draw the graph using pygraphviz for publication-quality figures.
 |
 |      Note that the style may not match one-to-one with the defined none/edge styles.
 |
 |      Parameters
 |      ----------
 |      filename : str, optional
 |          Name to save the figure to (if saving the figure). The default is ''.
 |      filetype : str, optional
 |          Type of file to safe. The default is 'png'.
 |      **kwargs : kwargs
 |          kwargs to draw.
 |
 |      Returns
 |      -------
 |      dot : PyGraphviz DiGraph
 |          Graph object corresponding to the figure.
 |
 |  get_dynamicnodes(self, mdl)
 |      Get dynamic node information for set_exec_order.
 |
 |  get_multi_edges(self, mdl, subedges)
 |      Attach functions/flows (subedges arg) to edges.
 |
 |      Parameters
 |      ----------
 |      mdl: Model
 |          Model to represent
 |      subedges : list
 |          nodes from the full graph which will become edges in the subgraph
 |          (e.g., individual flows)
 |
 |      Returns
 |      -------
 |      flows : dict
 |              Dictionary of edges with keys representing each sub-attribute of the
 |              edge (e.g., flows)
 |
 |  get_staticnodes(self, mdl)
 |      Get static node information for set_exec_order.
 |
 |  nx_from_obj(self, mdl)
 |      Generate the networkx.graph object corresponding to the model.
 |
 |      Parameters
 |      ----------
 |      mdl: Model
 |          Model to create the graph representation of
 |
 |      Returns
 |      -------
 |      g : networkx.Graph
 |          networkx.Graph representation of model functions and flows
 |          (along with their attributes)
 |
 |  set_exec_order(self, mdl, static={}, dynamic={}, next_edges={}, label_order=True, label_tstep=True)
 |      Overlay FunctionArchitectureGraph execution order data on graph structure.
 |
 |      Parameters
 |      ----------
 |      mdl : Model
 |          Model to plot the execution order of.
 |      static : dict/False, optional
 |          kwargs to overwrite the default style for functions/flows in the static
 |          execution step.
 |          If False, static functions are not differentiated. The default is {}.
 |      dynamic : dict/False, optional
 |          kwargs to overwrite the default style for functions/flows in the dynamic
 |          execution step.
 |          If False, dynamic functions are not differentiated. The default is {}.
 |      next_edges : dict
 |          kwargs to overwrite the default style for edges indicating the flow order.
 |          If False, these edges are not added. the default is {}.
 |      label_order : bool, optional
 |          Whether to label execution order (with a number on each node).
 |          The default is True.
 |      label_tstep : bool, optional
 |          Whether to label each timestep (with a number in the subtitle).
 |          The default is True.
 |
 |  set_flow_nodestates(self, mdl)
 |      Attaches node state attributes to Graph corresponding to the states of the
 |      model that belong to flows.
 |
 |      Parameters
 |      ----------
 |      mdl: Model
 |          Model to represent
 |
 |  set_fxn_nodestates(self, mdl)
 |      Attaches state attributes to Graph corresponding to the states of the model
 |      that belong to functions
 |
 |      Parameters
 |      ----------
 |      mdl: Model
 |          Model to represent
 |      time: float
 |          Time to execute indicators at. Default is 0.0
 |
 |  set_nx_states(self, mdl)
 |      Attach state attributes to Graph corresponding to the states of the model.
 |
 |      Parameters
 |      ----------
 |      mdl: Model
 |          Model to represent.
 |
 |  ----------------------------------------------------------------------
 |  Methods inherited from Graph:
 |
 |  add_node_groups(self, **node_groups)
 |      Create arbitrary groups of nodes to displayed with different styles.
 |
 |      Parameters
 |      ----------
 |      **node_groups : iterable
 |          nodes in groups. see example.
 |
 |      e.g.,::
 |      graph.add_node_groups(group1=('node1', 'node2'), group2=('node3'))
 |      graph.set_node_styles(group={'group1':{'color':'green'},
 |                                   'group2':{'color':'red'}})
 |      graph.draw()
 |
 |      would show two different groups of nodes, one with green nodes, and the other
 |      with red nodes
 |
 |  animate(self, history, times='all', figsize=(6, 4), **kwargs)
 |      Successively animate a plot using Graph.draw_from.
 |
 |      Parameters
 |      ----------
 |      history : History
 |          History with faulty and nominal states
 |      times : list, optional
 |          List of times to animate over. The default is 'all'
 |      figsize : tuple, optional
 |          Size for the figure. The default is (6,4)
 |      **kwargs : kwargs
 |
 |      Returns
 |      -------
 |      ani : matplotlib.animation.FuncAnimation
 |          Animation object with the given frames
 |
 |  calc_aspl(self)
 |      Compute average shortest path length of
 |
 |      Returns
 |      -------
 |      aspl: float
 |          Average shortest path length
 |
 |  calc_modularity(self)
 |      Compute network modularity of the graph.
 |
 |      Returns
 |      -------
 |      modularity : Modularity
 |
 |  calc_robustness_coefficient(self, trials=100, seed=False)
 |      Compute robustness coefficient of graph representation of model mdl.
 |
 |      Parameters
 |      ----------
 |      trials : int
 |          number of times to run robustness coefficient algorithm
 |          (result is averaged over all trials)
 |      seed : int
 |          optional seed to instantiate test with
 |
 |      Returns
 |      -------
 |      RC : robustness coefficient
 |
 |  draw(self, figsize=(12, 10), title='', fig=False, ax=False, withlegend=True, legend_bbox=(1, 0.5), legend_loc='center left', legend_labelspacing=2, legend_borderpad=1, **kwargs)
 |      Draw a graph with given styles corresponding to the node/edge properties.
 |
 |      Parameters
 |      ----------
 |      figsize : tuple, optional
 |          Size for the figure (plt.figure arg). The default is (12,10).
 |      title : str, optional
 |          Title for the plot. The default is "".
 |      fig : bool, optional
 |          matplotlib figure to project on (if provided). The default is False.
 |      ax : bool, optional
 |          matplotlib axis to plot on (if provided). The default is False.
 |      withlegend : bool, optional
 |          Whether to include a legend. The default is True.
 |      legend_bbox : tuple, optional
 |          bbox to anchor the legend to. The default is (1,0.5) (places legend on the right).
 |      legend_loc : str, optional
 |          loc argument for plt.legend. The default is "center left".
 |      legend_labelspacing : float, optional
 |          labelspacing argument for plt.legend. the default is 2.
 |      legend_borderpad : str, optional
 |          borderpad argument for plt.legend. the default is 1.
 |      **kwargs : kwargs
 |          Arguments for various supporting functions:
 |          (set_pos, set_edge_styles, set_edge_labels, set_node_styles,
 |          set_node_labels, etc)
 |
 |      Returns
 |      -------
 |      fig : matplotlib figure
 |          matplotlib figure to draw
 |      ax : matplotlib axis
 |          Ax in the figure
 |
 |  draw_from(self, time, history=, **kwargs)
 |      Draws the graph with degraded/fault data at a given time.
 |
 |      Parameters
 |      ----------
 |      time : int
 |          Time to draw the graph (in the history)
 |      history : History, optional
 |          History with nominal and faulty history. The default is History().
 |      **kwargs : **kwargs
 |          arguments for Graph.draw
 |
 |      Returns
 |      -------
 |      fig : matplotlib figure
 |          matplotlib figure to draw
 |      ax : matplotlib axis
 |          Ax in the figure
 |
 |  find_bridging_nodes(self)
 |      Determine bridging nodes in a graph representation of model mdl.
 |
 |      Returns
 |      -------
 |      bridgingNodes : list of bridging nodes
 |
 |  find_high_degree_nodes(self, p=90)
 |      Determine highest degree nodes, up to percentile p, in graph.
 |
 |      Parameters
 |      ----------
 |      p : int (optional)
 |          percentile of degrees to return, between 0 and 100
 |
 |      Returns
 |      -------
 |      highDegreeNodes : list of high degree nodes in format (node,degree)
 |
 |  get_obj_mode(self, obj)
 |
 |  get_obj_state(self, obj)
 |
 |  move_nodes(self, **kwargs)
 |      Set the position of nodes for plots in analyze.graph using a graphical tool.
 |
 |      Note: make sure matplotlib is set to plot in an external window
 |      (e.g., using '%matplotlib qt)
 |
 |      Parameters
 |      ----------
 |      **kwargs : kwargs
 |          keyword arguments for graph.draw
 |
 |      Returns
 |      -------
 |      p : GraphIterator
 |          Graph Iterator (in analyze.Graph)
 |
 |  plot_bridging_nodes(self, title='bridging nodes', node_kwargs={'node_color': 'red'}, **kwargs)
 |      Plot bridging nodes using self.draw().
 |
 |      Parameters
 |      ----------
 |      title : str, optional
 |          Title for the plot. The default is 'bridging nodes'.
 |      node_kwargs : dict, optional
 |          Non-default fields for NodeStyle
 |      **kwargs : kwargs
 |          kwargs for Graph.draw
 |
 |      Returns
 |      -------
 |      fig : matplotlib figure
 |          Figure
 |
 |  plot_degree_dist(self)
 |      Plots degree distribution of graph representation of model mdl.
 |
 |      Returns
 |      -------
 |      fig : matplotlib figure
 |          plot of distribution
 |
 |  plot_high_degree_nodes(self, p=90, title='', node_kwargs={'node_color': 'red'}, **kwargs)
 |      Plot high-degree nodes using self.draw()
 |
 |      Parameters
 |      ----------
 |      p : int (optional)
 |          percentile of degrees to return, between 0 and 100
 |      title : str, optional
 |          Title for the plot. The default is 'High Degree Nodes'.
 |      node_kwargs : dict : kwargs to overwrite the default NodeStyle
 |      **kwargs : kwargs
 |          kwargs for Graph.draw
 |
 |      Returns
 |      -------
 |      fig : matplotlib figure
 |          Figure
 |
 |  set_degraded(self, other)
 |      Set 'degraded' state in networkx graph.
 |
 |      Uses difference between states with another Graph object.
 |
 |      Parameters
 |      ----------
 |      other : Graph
 |          (assumed nominal) Graph to compare to
 |
 |  set_edge_labels(self, title='label', title2='', subtext='states', **edge_label_styles)
 |      Create labels using Labels.from_iterator for the edges in the graph.
 |
 |      Parameters
 |      ----------
 |      title : str, optional
 |          property to get for title text. The default is 'id'.
 |      title2 : str, optional
 |          property to get for title text after the colon. The default is ''.
 |      subtext : str, optional
 |          property to get for the subtext. The default is 'states'.
 |      **edge_label_styles : dict
 |          LabelStyle arguments to overwrite.
 |
 |  set_edge_styles(self, **edge_styles)
 |      Set self.edge_styles and self.edge_groups given the provided edge styles.
 |
 |      Parameters
 |      ----------
 |      **edge_styles : dict, optional
 |          Dictionary of tags, labels, and styles for the edges that overwrite the
 |          default. Has structure {tag:{label:kwargs}}, where kwargs are the keyword
 |          arguments to nx.draw_networkx_edges. The default is {"label":{}}.
 |
 |  set_heatmap(self, heatmap, cmap=<matplotlib.colors.LinearSegmentedColormap object at 0x000002076266ECE0>, default_color_val=0.0)
 |      Set the association and plotting of a heatmap on a graph.
 |
 |      e.g.,::
 |      graph.set_heatmap({'node_1':1.0, 'node_2': 0.0, 'node_3':0.5})
 |      graph.draw()
 |
 |      Should draw node_1 the bluest, node_2 the reddest, and node_3 in between.
 |
 |      Parameters
 |      ----------
 |      heatmap : dict/result
 |          dict/result with keys corresponding to the nodes and values in the range
 |          of a heatmap (0-1)
 |      cmap : mpl.Colormap, optional
 |          Colormap to use for the heatmap. The default is plt.cm.coolwarm.
 |      default_color_val : float, optional
 |          Value to use if a node is not in the heatmap dict. The default is 0.0.
 |
 |  set_node_labels(self, title='id', title2='', subtext='', **node_label_styles)
 |      Create labels using Labels.from_iterator for the nodes in the graph.
 |
 |      Parameters
 |      ----------
 |      title : str, optional
 |          Property to get for title text. The default is ‘id’.
 |      title2 : str, optional
 |          Property to get for title text after the colon. The default is ‘’.
 |      subtext : str, optional
 |          property to get for the subtext. The default is ‘’.
 |      node_label_styles :  dict
 |          LabelStyle arguments to overwrite.
 |
 |  set_node_styles(self, **node_styles)
 |      Set self.node_styles and self.edge_groups given the provided node styles.
 |
 |      Parameters
 |      ----------
 |      **node_styles : dict, optional
 |          Dictionary of tags, labels, and style kwargs for the nodes that overwrite
 |          the default. Has structure {tag:{label:kwargs}}, where kwargs are the
 |          keyword arguments to nx.draw_networkx_nodes. The default is {"label":{}}.
 |
 |  set_pos(self, auto=True, **pos)
 |      Set graph positions to given positions, (automatically or manually).
 |
 |      Parameters
 |      ----------
 |      auto : str, optional
 |          Whether to auto-layout the node position. The default is True.
 |      **pos : nodename=(x,y)
 |          Positions of nodes to set. Otherwise updates to the auto-layout or (0.5,0.5)
 |
 |  set_resgraph(self, other=False)
 |      Process results for results graphs (show faults and degradations).
 |
 |      Parameters
 |      ----------
 |      other : Graph, optional
 |          Graph to compare with (for degradations). The default is False.
 |
 |  sff_model(self, endtime=5, pi=0.1, pr=0.1, num_trials=100, start_node='random', error_bar_option='off')
 |      susc-fix-fail model.
 |
 |      Parameters
 |      ----------
 |      endtime: int
 |          simulation end time
 |      pi : float
 |          infection (failure spread) rate
 |      pr : float
 |          recovery (fix) rate
 |      num_trials : int
 |          number of times to run the epidemic model, default is 100
 |      error_bar_option : str
 |          option for plotting error bars (first to third quartile), default is off
 |      start_node : str
 |          start node to use in the trial. default is 'random'
 |
 |      Returns
 |      -------
 |      fig: plot of susc, fail, and fix nodes over time
 |
 |  ----------------------------------------------------------------------
 |  Data descriptors inherited from Graph:
 |
 |  __dict__
 |      dictionary for instance variables (if defined)
 |
 |  __weakref__
 |      list of weak references to the object (if defined)

The main function used for displaying model structure is draw()

help(an.graph.FunctionArchitectureGraph.draw)

Help on function draw in module fmdtools.analyze.graph:

draw(self, figsize=(12, 10), title='', fig=False, ax=False, withlegend=True, legend_bbox=(1, 0.5), legend_loc='center left', legend_labelspacing=2, legend_borderpad=1, **kwargs)
    Draw a graph with given styles corresponding to the node/edge properties.

    Parameters
    ----------
    figsize : tuple, optional
        Size for the figure (plt.figure arg). The default is (12,10).
    title : str, optional
        Title for the plot. The default is "".
    fig : bool, optional
        matplotlib figure to project on (if provided). The default is False.
    ax : bool, optional
        matplotlib axis to plot on (if provided). The default is False.
    withlegend : bool, optional
        Whether to include a legend. The default is True.
    legend_bbox : tuple, optional
        bbox to anchor the legend to. The default is (1,0.5) (places legend on the right).
    legend_loc : str, optional
        loc argument for plt.legend. The default is "center left".
    legend_labelspacing : float, optional
        labelspacing argument for plt.legend. the default is 2.
    legend_borderpad : str, optional
        borderpad argument for plt.legend. the default is 1.
    **kwargs : kwargs
        Arguments for various supporting functions:
        (set_pos, set_edge_styles, set_edge_labels, set_node_styles,
        set_node_labels, etc)

    Returns
    -------
    fig : matplotlib figure
        matplotlib figure to draw
    ax : matplotlib axis
        Ax in the figure

However, these relationships only define the structure of the model–for the model to simulate correctly and efficiently, the run order of fuctions additionally must be defined.

Each timestep of a model simulation can be broken down into two steps:

• Dynamic Propagation Step: In the dynamic propagation step, the time-based behaviors (e.g. accumulations, movement, etc.) are each run once in a specified order. These steps are generally quicker to execute because each behavior is only run once and the flows do not need to be tracked to determine which behaviors to execute next. This step is run first at each time-step of the model.

• Static Propagation Step: In the static propagation step, behaviors are propagated between functions iteratively until the state of the model converges to a single value. This may require an update of multiple function behaviors until there are no more new behaviors to run. Thus, static behaviors should be ‘’timeless’’ (always give the same output for the same input) and convergent (behaviors in each function should not change each other ad infinitum). This step is run second at each time-step of the model.

With these different behaviors, one can express a range of different types of models: - static models where only only one timestep is run, where fault scenarios show the immediate propagation of faults through the system. - dynamic models where a number of timesteps are run (but behaviors are only run once). - hybrid models where dynamic behaviors are run once and then a static propagation step is performed at each time-step.

The main interfaces/functions involved in defining run order are: - Function.static_behavior(self, time) and Function.behavior(self, time), which define function behaviors which occur during the static propagation step. - Function.dynamic_behavior(self, time), which defines function behaviors during the dynamic propagation step. - Function.condfaults(self, time), does not define run order, but it can be used for behaviors which would at any time during simulation (dynamic or static) lead to the system entering a fault mode (e.g., conditional faults). - FunctionArchitecture.add_fxn(), which when used successively for each function specifies that those functions run in the order they are added.

The overall static/dynamic propagation steps of the model can then be visualized using ModelGraph.set_exec_order successively with ModelGraph.draw().

help(an.graph.FunctionArchitectureGraph.set_exec_order)

Help on function set_exec_order in module fmdtools.analyze.graph:

set_exec_order(self, mdl, static={}, dynamic={}, next_edges={}, label_order=True, label_tstep=True)
    Overlay FunctionArchitectureGraph execution order data on graph structure.

    Parameters
    ----------
    mdl : Model
        Model to plot the execution order of.
    static : dict/False, optional
        kwargs to overwrite the default style for functions/flows in the static
        execution step.
        If False, static functions are not differentiated. The default is {}.
    dynamic : dict/False, optional
        kwargs to overwrite the default style for functions/flows in the dynamic
        execution step.
        If False, dynamic functions are not differentiated. The default is {}.
    next_edges : dict
        kwargs to overwrite the default style for edges indicating the flow order.
        If False, these edges are not added. the default is {}.
    label_order : bool, optional
        Whether to label execution order (with a number on each node).
        The default is True.
    label_tstep : bool, optional
        Whether to label each timestep (with a number in the subtitle).
        The default is True.

The next sections will demonstrate these functions using a simple hybrid model.

Model Setup

Consider the following (highly simplified) rover electrical/navigation model. We can define the functions of this rover using the classes:

from fmdtools.define.container.state import State
from fmdtools.define.flow.base import Flow
from fmdtools.define.container.mode import Mode
class ControlState(State):
    power: float=0.0
    vel:   float=0.0
class Control(Flow):
    __slots__=()
    container_s = ControlState
class ControlRoverMode(Mode):
    fm_args={'no_con':(1e-4, 200)}
    opermodes = ('drive', 'standby')
    mode : str='standby'

class ControlRover(Function):
    __slots__=('control',)
    container_m = ControlRoverMode
    flow_control = Control
    def dynamic_behavior(self,time):
        if not self.m.in_mode('no_con'):
            if time == 5:  self.m.set_mode('drive')
            if time == 50: self.m.set_mode('standby')
        if self.m.in_mode('drive'):
            self.control.s.power = 1.0
            self.control.s.vel = 1.0
        elif self.m.in_mode('standby'):
            self.control.s.vel = 0.0
            self.control.s.power=0.0

This function uses dynamic_behavior() to define the dynamic behavior of going through different modes depending on what model time it is. While this could also be entered in as a static behavior, because none of the defined behaviors themselves result from external inputs, there is no reason to.

class ForceState(State):
    transfer:  float=1.0
    magnitude: float=1.0
class Force(Flow):
    __slots__ = ()
    container_s = ForceState

class EEState(State):
    v:  float=0.0
    a:  float=0.0
class EE(Flow):
    __slots__ = ()
    container_s = EEState
class GroundState(State):
    x:  float=0.0
class Ground(Flow):
    __slots__=()
    container_s = GroundState

class MoveRoverMode(Mode):
    fm_args={"mech_loss": (1.0,), "short": (1.0,), "elec_open": (1.0,)}
class MoveRoverState(State):
    power: float=0.0

class MoveRover(Function):
    __slots__ = ('ee', 'control', 'ground', 'force')
    container_s = MoveRoverState
    container_m = MoveRoverMode
    flow_ee = EE
    flow_control = Control
    flow_ground = Ground
    flow_force = Force
    def static_behavior(self, time):
        self.s.power = self.ee.s.v * self.control.s.vel *self.m.no_fault("elec_open")
        self.ee.s.a = self.s.power/(12*(self.m.no_fault('short')+0.001))
        if self.s.power >100: self.m.add_fault("elec_open")
    def dynamic_behavior(self, time):
        if not self.m.has_fault("elec_open", "mech_loss"):
            self.ground.s.x = self.ground.s.x + self.s.power*self.m.no_fault("mech_loss")

The Move_Rover function uses both: - a static behavior which defines the input/output of electrical power at each instant, and - a dynamic behavior which defines the movement of the rover over time

In this instance, the static behavior is important for enabling faults to propagate instantaneously in a single time-step (in this case, a short causing high current load to the battery).

from fmdtools.define.container.time import Time

class StoreEnergyState(State):
    charge: float=100.0
class StoreEnergyTime(Time):
    local_dt = 0.5
class StoreEnergyMode(Mode):
    fm_args = {"no_charge":(1e-5, 100, {'standby':1.0}),
                   "short":(1e-5, 100, {'supply':1.0})}
    opermodes = ("supply","charge","standby")
    exclusive = True
    key_phases_by = "self"
    mode: str = "standby"

class StoreEnergy(Function):
    __slots__ = ('ee', 'control')
    container_s = StoreEnergyState
    container_t = StoreEnergyTime
    container_m = StoreEnergyMode
    flow_ee = EE
    flow_control = Control
    def static_behavior(self,time):
        if self.ee.s.a > 5: self.m.add_fault("no_charge")
    def dynamic_behavior(self,time):
        if self.m.in_mode("standby"):
            self.ee.s.put(v = 0.0, a=0.0)
            if self.control.s.power==1: self.m.set_mode("supply")
        elif self.m.in_mode("charge"):
            self.s.charge =min(self.s.charge+self.t.dt, 20.0)
        elif self.m.in_mode("supply"):
            if self.s.charge > 0:
                self.ee.s.v = 12.0
                self.s.charge -= self.t.dt
            else: self.m.set_mode("no_charge")
            if self.control.s.power==0:
                self.m.set_mode("standby")
        elif self.m.in_mode("short"):
            self.ee.s.v = 100.0
            self.s.charge = 0.0
        elif self.m.in_mode("no_charge"):
            self.ee.s.v=0.0
            self.s.charge = 0.0

The Store_Energy function has both a static behavior and a dynamic behavior. In this case, the static behavior enables the propagation of an adverse current from the drive system to damage the battery instantaneously, (instead of over several timesteps).

We’ve also set a local timestep dt=0.5, which simulates the ‘dynamic_behavior’ twice as often as fxngraph. This could be used to enable higher-grained behaviors for dttstep–in this case, the expected behaviors are not expected to change.

class VideoState(State):
    line:  float=0.0
    angle: float=0.0
class Video(Flow):
    __slots__=()
    container_s = VideoState

class ViewGround(Function):
    __slots__=('ground', 'ee', 'video', 'force')
    flow_ground = Ground
    flow_ee = EE
    flow_video = Video
    flow_force = Force

class Communicate(Function):
    __slots__ = ('comms', 'ee')
    flow_comms = Flow
    flow_ee = EE

class Rover(FunctionArchitecture):
    default_sp = {'times':(0,60), 'phases':(('start',1,30), ('end',31, 60))}
    def init_architecture(self, **kwargs):
        self.add_flow('ground', Ground)
        self.add_flow('force', Force)
        self.add_flow('ee', EE)
        self.add_flow('video', Video)
        self.add_flow('control', Control)
        self.add_flow('comms', Flow) #{'x':0,'y':0}

        self.add_fxn("control_rover", ControlRover, "control")
        self.add_fxn("store_energy", StoreEnergy,  "ee", "control")
        self.add_fxn("move_rover", MoveRover, "ground","ee", "control", "force")
        self.add_fxn("view_ground", ViewGround, "ground", "ee", "video","force")
        self.add_fxn("communicate", Communicate, "comms", "ee")

rover_pos = {'control_rover': [-0.017014983401385075, 0.8197778602536954],
 'move_rover': [0.1943738434915952, -0.5118219332727401],
 'store_energy': [-0.256309000069049, -0.004117688709924516],
 'view_ground': [-0.7869889764273651, 0.47147713497270827],
 'communicate': [0.5107674237596388, 0.4117119127760298],
 'ground': [-0.7803536309752367, -0.4502200140852195],
 'force': [0.4327741966569625, 0.13966361395868865],
 'ee': [-0.6981138376424448, 0.13829658866345518],
 'video': [-0.49486453723245205, 0.698244546263499],
 'control': [0.11615283552311584, -0.1842023746850714],
 'comms': [0.3373143873188402, 0.6507526319915691]}

Graph Visualization

Without defining anything about the simulation itself, the containment relationships between the model structures can be visualized using FunctionArchitectureGraph, FunctionArchitectureFlowGraph, FunctionArchitectureFxnGraph, and FunctionArchitectureTypeGraph.

from fmdtools.analyze.graph.architecture import FunctionArchitectureGraph, FunctionArchitectureFlowGraph
from fmdtools.analyze.graph.architecture import FunctionArchitectureFxnGraph, FunctionArchitectureTypeGraph

mdl = Rover()
mtg = FunctionArchitectureTypeGraph(mdl)
fig, ax = mtg.draw()

As shown, because the class for view ground and communicate are undefined, they are shown here as both instantiations of the Function class, which does not simulate. Additionally, Force, Ground, Video and Flow (used for comms) are left dangling since they aren’t actually connected to anything.

This same structure can also be visualized using the graphviz renderer, as shown below.

dot = mtg.draw_graphviz()

By default, draw uses matplotlib via networkx’s built-in plotting functions Other renderers are considered experimental and may not support every interface style argument without bugs.

However, graphviz is much more fully-featured and often creates nicer-looking plots. The reason it is not used by default (and why Graph was not build around it) is that graphviz needs to be installed externally.

Graph Views

While the FunctionArchitectureTypeGraph view shows the containtment relationships of the classes, the structural relationships between the model functions and flows can be viewed using the FunctionArchitectureFxnGraph, FunctionArchitectureFlowGraph and FunctionArchitectureGraph classes, shown below.

mfg = FunctionArchitectureFxnGraph(mdl)
fig, ax = mfg.draw()

[i.get_label() for i in ax.get_legend().legendHandles]

C:\Users\dhulse\AppData\Local\Temp\1\ipykernel_1324\1233849047.py:1: MatplotlibDeprecationWarning: The legendHandles attribute was deprecated in Matplotlib 3.7 and will be removed two minor releases later. Use legend_handles instead.
  [i.get_label() for i in ax.get_legend().legendHandles]

['Function', 'flows']

mfg.edge_labels

Labels(title={('control_rover', 'store_energy'): '<flows>', ('control_rover', 'move_rover'): '<flows>', ('store_energy', 'communicate'): '<flows>', ('store_energy', 'view_ground'): '<flows>', ('store_energy', 'move_rover'): '<flows>', ('move_rover', 'communicate'): '<flows>', ('move_rover', 'view_ground'): '<flows>', ('view_ground', 'communicate'): '<flows>'}, title_style=EdgeLabelStyle(font_size=12, font_color='k', font_weight='normal', alpha=1.0, horizontalalignment='center', verticalalignment='bottom', clip_on=False, bbox={'alpha': 0}, rotate=False), subtext={('control_rover', 'store_energy'): "['control']", ('control_rover', 'move_rover'): "['control']", ('store_energy', 'communicate'): "['ee']", ('store_energy', 'view_ground'): "['ee']", ('store_energy', 'move_rover'): "['control', 'ee']", ('move_rover', 'communicate'): "['ee']", ('move_rover', 'view_ground'): "['force', 'ee', 'ground']", ('view_ground', 'communicate'): "['ee']"}, subtext_style=EdgeLabelStyle(font_size=12, font_color='k', font_weight='normal', alpha=1.0, horizontalalignment='center', verticalalignment='top', clip_on=False, bbox={'alpha': 0}, rotate=False))

Note the limitations with this representation. Specifically, flows mapped onto edges may duplicated each other (notice, for example, how many edges are listed with the ee flow! This - makes it difficult to visualize how multiple nodes are connected through the same flow - makes it difficult to label edges (since each edge may have a number of flows on it) - leads to many edge overlaps

The FunctionArchitectureFlowGraph is similarly limited:

mfg = FunctionArchitectureFlowGraph(mdl)
fig, ax = mfg.draw()

As a result, we usually use the standard FunctionArchitectureGraph class, which captures the full bipartite structure of models–where both functions and flows are nodes in the graph.

mg = FunctionArchitectureGraph(mdl)
mg.set_pos(**rover_pos)
fig, ax = mg.draw()

Note that these types are also compatible with graphviz. Graphviz output can be customized using options for the renderer (see http://www.graphviz.org/doc/info/attrs.html for all options)

For example:

dot = mg.draw_graphviz(layout='neato', overlap="voronoi")

Run Order

To specify the run order of this model, the add_fxn method is used. The order of the call defines the run order of each instantiated function (Control_Rover -> Move_Rover -> Store_Energy -> View_Ground -> Communicate_Externally).

In addition to the functions which have been defined here, this model additionally has a number of functions which have not been defined (and will thus not execute). We can overlay these execution characteristics onto the graph using FunctionArchitectureGraph.set_exec_order.

help(FunctionArchitectureGraph.set_exec_order)

Help on function set_exec_order in module fmdtools.analyze.graph:

set_exec_order(self, mdl, static={}, dynamic={}, next_edges={}, label_order=True, label_tstep=True)
    Overlay FunctionArchitectureGraph execution order data on graph structure.

    Parameters
    ----------
    mdl : Model
        Model to plot the execution order of.
    static : dict/False, optional
        kwargs to overwrite the default style for functions/flows in the static
        execution step.
        If False, static functions are not differentiated. The default is {}.
    dynamic : dict/False, optional
        kwargs to overwrite the default style for functions/flows in the dynamic
        execution step.
        If False, dynamic functions are not differentiated. The default is {}.
    next_edges : dict
        kwargs to overwrite the default style for edges indicating the flow order.
        If False, these edges are not added. the default is {}.
    label_order : bool, optional
        Whether to label execution order (with a number on each node).
        The default is True.
    label_tstep : bool, optional
        Whether to label each timestep (with a number in the subtitle).
        The default is True.

mg.set_exec_order(mdl)
fig, ax = mg.draw()

As shown, functions and flows active in the static propagation step are highlighted in cyan while the functions in the dynamic propagation step are shown (or given a border) in teal. Functions without behaviors are shown in light grey, and the run order of the dynamic propagation step is shown as numbers under the corresponding functions.

Additionally, by default the local timestep dt is labelled so we can see if/how it deviates from the overall model timestep–see, for example, the Store_Energy function.

In addition, Model.plot_dynamic_run_order can be used to visualize the dynamic propagation step.

mdl.plot_dynamic_run_order()

(<Figure size 640x480 with 1 Axes>, <Axes: >)

This plot shows that the dynamic execution step runs in the order defined in the Model module: first, Control_Rover, then, Store_Energy, and finally Move_Rover (reading left-to-right in the upper axis). The plot additionally shows which flows correspond to these function as it progresses through execution, which enables some understanding of which data structures are used or acted on at each execution time.

Behavior/Fault Visualization

To verify the static propagation of the short mode in the move_rover function, we can view the results of that scenario. As was set up, the intention of using the static propagation step was to enable the resulting fault behavior (a spike in current followed by a loss of charge) to occur in a single timestep.

In general, one can plot the effects of faults over time, using methods in plot, as shown below.

result, mdlhist = prop.one_fault(mdl, "move_rover", "short", 10, desired_result='graph')

result.graph

<fmdtools.analyze.graph.FunctionArchitectureGraph at 0x2070e98a860>

fig, axs = mdlhist.plot_line(
                            'flows.ee.s.v',
                            'flows.ee.s.a',
                            'flows.control.s.power',
                            'flows.control.s.vel',
                            'fxns.store_energy.s.charge',
                            'fxns.store_energy.m.mode',
                            'fxns.move_rover.s.power', time_slice=10)

As shown, the static propagation step enables the mode to propagate back to the store_energy function (causing the no_charge fault) in the same timestep it is injected, even though it occurs later in the propagation order.

However, because the voltage and current output behaviors for the function are defined in the dynamic_behavior method of the store_energy function, these are only updated to their final value (of zero) at the next step. While this enables some visualization of the current spike, it may keep faults and behaviors from further propagating through the functions as desired. Thus, to enable this, one might reallocate some of the behaviors from the dynamic_behavior method to the static_behavior method.

Visualizing time-slices

Contained in the result output in propagate.one_fault is the graph specified in desired_result, which is a graph (multiple types may be provided). When output from propagate the Resgraph is by default given state information for each function/flow, as well as degraded and faulty properties which correlate with whether the State values deviate from those in the nominal.

Note that by default there are three main classifications for functions/flows visualized in this type of plot: - red: faulty function. Notes that the function is in a fault mode - orange: degraded function/flow. Nodes that the values of the flow or states of the function are different from the nominal scenario. Note that this is different than saying the values represent a problem, since contingency actions are also different between the nominal and faulty runs. - grey: nominal function/flow. This notes that there is nothing different between the nominal and faulty run in that function or flow.

These styles may be changed at will.

result.graph.set_pos(**rover_pos)
fig, ax = result.graph.draw()

dot = result.graph.draw_graphviz()

The resgraph only gives the state of the model at the final state. We might instead want to visualize the graph at different given times. This is performed with Graph.draw_from:

help(FunctionArchitectureGraph.draw_from)

Help on function draw_from in module fmdtools.analyze.graph:

draw_from(self, time, history=, **kwargs)
    Draws the graph with degraded/fault data at a given time.

    Parameters
    ----------
    time : int
        Time to draw the graph (in the history)
    history : History, optional
        History with nominal and faulty history. The default is History().
    **kwargs : **kwargs
        arguments for Graph.draw

    Returns
    -------
    fig : matplotlib figure
        matplotlib figure to draw
    ax : matplotlib axis
        Ax in the figure

rmg = FunctionArchitectureGraph(mdl)
rmg.set_pos(**rover_pos)
fig, ax = rmg.draw_from(5, mdlhist)

fig, ax = rmg.draw_from(10, mdlhist)

fig, ax = rmg.draw_from(25, mdlhist)

This can can be done automatically over a number of different times using Graph.animate:

help(FunctionArchitectureGraph.animate)

Help on function animate in module fmdtools.analyze.graph:

animate(self, history, times='all', figsize=(6, 4), **kwargs)
    Successively animate a plot using Graph.draw_from.

    Parameters
    ----------
    history : History
        History with faulty and nominal states
    times : list, optional
        List of times to animate over. The default is 'all'
    figsize : tuple, optional
        Size for the figure. The default is (6,4)
    **kwargs : kwargs

    Returns
    -------
    ani : matplotlib.animation.FuncAnimation
        Animation object with the given frames

ani = rmg.animate(mdlhist)

from IPython.display import HTML
HTML(ani.to_jshtml())

Once Loop Reflect